package intensivecomp.mapred;

import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;
import java.util.StringTokenizer;

public class Map extends Mapper<Object,Text,Text,DoubleWritable> {
    public void map(Object key,Text value,Context context) throws IOException,InterruptedException{
        //对输入的行切词
        String[] points = value.toString().split("\n");
        Point p1 = getPoint(points[0]), p2 = getPoint(points[1]);
        DoubleWritable v = new DoubleWritable();
        v.set(p1.GetDistance(p2));
        Text k = new Text();
        k.set(key.toString());
        context.write(k, v);
    }

    public Point getPoint(String record){
        String[] info = record.split(",");
        double lat = Double.parseDouble(info[0]);
        double lng = Double.parseDouble(info[1]);
        return new Point(lng,lat);
    }

    class Point{
        double lng, lat;

        private double EARTH_RADIUS = 6378.137;

        private double rad(double d) {
            return d * Math.PI / 180.0;
        }

        public Point(double lng, double lat) {
            this.lng = lng;
            this.lat = lat;
        }

        public double GetDistance(Point p2) {
            double lat2 = p2.lat;
            double lng2 = p2.lng;
            double radLat1 = rad(lat);
            double radLat2 = rad(lat2);
            double a = radLat1 - radLat2;
            double b = rad(lng) - rad(lng2);
            double s = 2 * Math.asin(Math.sqrt(Math.pow(Math.sin(a/2),2) +
                    Math.cos(radLat1)*Math.cos(radLat2)*Math.pow(Math.sin(b/2),2)));
            s = s * EARTH_RADIUS;
            s = Math.round(s * 1000);
            return s;
        }
    }
}
